package com.iot.service.mqtt.handler;

import com.iot.service.mqtt.config.Constant;
import com.iot.service.mqtt.model.fix.TlMqttFixedHead;
import com.iot.service.mqtt.model.request.TlMqttDisconnectReq;
import com.iot.service.mqtt.model.variable.TlMqttPublishVariableHead;
import com.iot.service.mqtt.retry.RetryStrategy;
import com.iot.service.mqtt.service.message.IMessage;
import com.iot.service.mqtt.service.pub.IPublishStoreService;
import com.iot.service.mqtt.service.rel.IPubrelStoreService;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.subscripe.ISubscriptionService;
import com.iot.service.mqtt.utils.ChannelManager;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.mqtt.MqttQoS;
import io.netty.util.AttributeKey;
import lombok.RequiredArgsConstructor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

/**
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
@RequiredArgsConstructor
public class TlMqttDisconnectHandler extends SimpleChannelInboundHandler<TlMqttDisconnectReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttDisconnectHandler.class);

    private final ISubscriptionService subscriptionService;

    private final ISessionService sessionService;

    private final ChannelManager channelManager;

    private final IPublishStoreService publishStoreService;

    private final IPubrelStoreService pubrelStoreService;


    private final IMessage messageIdServiceId;

    private final RetryStrategy retryStrategy;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttDisconnectReq msg) throws Exception {
        //断开标志位设置为true 这样就不发生遗嘱消息了
        ctx.channel().attr(AttributeKey.valueOf(Constant.DISCONNECT)).set(true);
        ctx.channel().close();
        logger.debug("disconnect...");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx)  {

        //判断是否是正常断开的
        Boolean disconnectFlag= (Boolean)ctx.channel().attr(AttributeKey.valueOf(Constant.DISCONNECT)).get();
        //判定是否是mqtt连接
        Object clientIdObj = ctx.channel().attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if(null==clientIdObj){
            ctx.channel().close();
            return;
        }
        String clientId = clientIdObj.toString();
        logger.info("client 【{}】 disconnect",clientId);
        sessionService.find(clientId)
                .doOnSuccess(session ->{
                    //从channelManager中移除
                    channelManager.remove(clientId);
                    ctx.channel().close();
                } )
                .flatMap(session -> {

                    /*当CleanSession为0的会话断开后，服务器还必须将所有和客户端订阅相关的QoS1和QoS2的消息作为会话状态的一部分存储起来[MQTT-3.1.2-5]。*/
                    if(session.getCleanSession().equals(Boolean.TRUE)){
                        logger.debug("移除会话");
                        return actionOnCleanSession(clientId);
                    }else{
                        session.setOnline(false);
                        //如果规定时间内没有连接的话就移除这个会话
                      //  retryStrategy.removeSession(clientId);
                        return sessionService.save(session);
                    }
                }).subscribe(success -> {

                    // todo 起一个定时任务   n分钟后如何还没有连接 就直接断开
                    if(disconnectFlag){
                        //如果断开标志位是true 表示是正常断开的 就不发生遗嘱消息
                       publishStoreService.clearWillMessage(clientId).subscribe();
                       return;
                    }
                    logger.debug("发送遗嘱消息");
                    logger.debug("client 【{}】 disconnect unsafe send willMessage if has",clientId);
                    publishStoreService.findWillMessage(clientId)
                            .flatMap(willMsg -> {
                                TlMqttPublishVariableHead variableHead = willMsg.getVariableHead();
                                String topic = variableHead.getTopic();
                                TlMqttFixedHead fixedHead = willMsg.getFixedHead();
                                MqttQoS qos = fixedHead.getQos();
                                if(qos!=MqttQoS.AT_MOST_ONCE){
                                    Long messageId = messageIdServiceId.getMessageId();
                                    variableHead.setMessageId(messageId);
                                }
                                return subscriptionService.find(topic)
                                        // 向所有订阅者发送消息
                                        .doOnNext(sub ->
                                                channelManager.writeAndFlush(sub.getClientId(), willMsg)
                                        )
                                        // 发送完成后清理遗嘱消息
                                        .then(Mono.defer(() ->
                                                publishStoreService.clearWillMessage(clientId)
                                        ));
                            })
                            .subscribe(
                                    ignored -> {},
                                    error -> logger.error("处理遗嘱消息失败", error)
                            );
                    });
    }

    /**
     * @description: 清除客户端的信息
     * @author: hszhou
     * @datetime: 2025-04-28 16:32:19
     * @param:
     * @param: clientId 客户端ID
     * @return: void
     **/
    private Mono<Boolean> actionOnCleanSession(String clientId) {
        // sessionService.clear(String ClientId) 方法返回 true 则表明该 clientId 之前的 cleanSession = false, 那么应该继续清理
        // 订阅信息、pub 信息、 pubRel 信息, 否则无需清理
        return sessionService.clear(clientId).flatMap(e -> {
            if (e) {
                return Mono.when(
                        //清除所有的订阅
                        subscriptionService.clear(clientId),
                        //清除保存的消息
                        publishStoreService.clearPublishMessage(clientId),
                        pubrelStoreService.clearPubRelMessage(clientId),
                        //清除遗嘱消息
                        publishStoreService.clearWillMessage(clientId)).thenReturn(true);
            } else {
                return Mono.just(true);
            }
        });
    }
}
